package insurance.util

/**
 * @Package : com.bw.spark.insurance.util
 * @Description : TODO
 * @Create on : 2022/12/24 21:04
 * @author chuzhigao shanghai bawei hongyi company
 * @version
 */
import org.apache.commons.lang3.StringUtils
import org.apache.spark.sql.{DataFrame, SaveMode, SparkSession}
import java.io.{File, FileInputStream, InputStream}
import java.util.Properties
import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.fs.{FileSystem, Path}
import scala.io.Source

object SparkUtil {
    // 接收一个 SparkSession 对象 和 一个sql文件
  def executeSQLFile(spark: SparkSession, filename: String) = {
      //filename
      /*基础路径：资源应该位于你的项目的类路径（classpath）中。如果你使用的是Maven或Gradle，这通常意味着资源应该放在src/main/resources目录下。
      相对路径：你应该提供相对于类路径的相对路径。不需要在路径前面加上斜杠（/）。
      文件名：直接提供文件名，如果文件位于类路径的根目录下。*/
      val in: InputStream = this.getClass.getClassLoader().getResourceAsStream(filename);
      val sqls: String = Source.fromInputStream(in).mkString //mkString将所有行连接成一个字符串sql

    sqls
      .split(";")
      .filter(StringUtils.isNoneBlank(_))
      .foreach(sql => {
        val start = System.currentTimeMillis()
        println(sql + ";")
        val df: DataFrame = spark.sql(sql)
        if (sql.split("\n").filter(!_.trim.startsWith("--")).mkString.trim.startsWith("select")) {
          df.show()
        }
        val end = System.currentTimeMillis()
        println("耗时:"+(end-start)/1000+"秒")
      })
  }

    def executeSQLFileFromHdfs(spark: SparkSession, filename: String,time_month: String,table_name:String) = {

        val conf = new Configuration()
        val fs = FileSystem.get(conf)

        val filePath = new Path(filename)

        val in: InputStream = fs.open(filePath);

        //oozie 调度一定要指定编码
        val sqls: String = Source.fromInputStream(in,"UTF-8").mkString
        println(time_month+">>>>>>>")
        spark.sql("set time_month="+time_month)
        val tt: String = "select '${time_month}'"
        spark.sql(tt).show()

        sqls
                .split(";")
                .filter(StringUtils.isNoneBlank(_))
                .foreach(sql => {
                    val start = System.currentTimeMillis()
                    println(sql + ";")
                    val df: DataFrame = spark.sql(sql)
                    if (sql.split("\n").filter(!_.trim.startsWith("--")).mkString.trim.startsWith("select")) {
                        df.show()
                    }
                    val end = System.currentTimeMillis()
                    println("耗时:" + (end - start) / 1000 + "秒")
                })


        if(table_name != null ){
            val result: DataFrame = spark.sql("select * from insurance_app." + table_name + " where month = substr(${time_month},0,7)")
            result.show(10)
            result
                    .write
                    .mode(SaveMode.Overwrite)
                    .jdbc(mysql_url, table_name, jdbc_prop)
        }



    }

  val mysql_url="jdbc:mysql://hadoop102:3306/insurance_app?createDatabaseIfNotExist=true&serverTimezone=UTC&characterEncoding=utf8&useUnicode=true"
  val jdbc_prop = new Properties()
  jdbc_prop.setProperty("user", "root")
  jdbc_prop.setProperty("password", "123456")
}
